#coding:utf8


from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StringType, IntegerType, StructField
import pyspark.sql.functions as F

if __name__ == '__main__':
    # 构建SparkSession执行环境入口对象
    spark = SparkSession.builder.\
        appName("test_parque").\
        master("local[*]"). \
            getOrCreate()
    sc = spark.sparkContext

    # 1. 从JDBC读取数据
    df = spark.read.format("jdbc"). \
        option('url', 'jdbc:mysql://localhost:3306/bigdata?useSSL=false&useUnicode=true'). \
        option('dbtable', 'movie_table'). \
        option('user', 'root'). \
        option('password', '123456').load()

    # 2.从文件读取数据
    # struct = StructType().add("user_id", StringType()). \
    #     add("movie_id", StringType()). \
    #     add("rank", IntegerType()). \
    #     add("create_time", IntegerType())
    # df = spark.read.format("csv"). \
    #     option("sep", "\t"). \
    #     option("header", False). \
    #     option("encoding", "utf-8"). \
    #     schema(schema=struct). \
    #     load('../data/input/u.data')
    # df.show()

    # 将数据写入到数据库中
    df.write.format('jdbc').\
        mode('append').\
        option('url','jdbc:mysql://localhost:3306/bigdata?useSSL=false&useUnicode=true').\
        option('dbtable','movie_table').\
        option('user','root').\
        option('password','123456').\
        save()